package com.kafka.study.config;

import org.apache.kafka.clients.producer.ProducerConfig;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.core.ProducerFactory;

import java.util.HashMap;
import java.util.Map;

@Configuration
public class PConfig {
    @Value("${spring.kafka.bootstrap-servers}")
    private String bootstrapServers;

    @Value("${spring.kafka.producer.key-serializer}")
    private String keySerializer;

    @Value("${spring.kafka.producer.value-serializer}")
    private String valueSerializer;

    @Value("${spring.kafka.producer.time.compression-type}")
    private String timeCompressionType;

    @Value("${spring.kafka.producer.time.linger-ms}")
    private String timeLingerMs;

    @Value("${spring.kafka.producer.time.batch-size}")
    private String timeBatchSize;

    @Value("${spring.kafka.producer.business.compression-type}")
    private String businessCompressionType;

    @Value("${spring.kafka.producer.business.linger-ms}")
    private String businessLingerMs;

    @Value("${spring.kafka.producer.business.batch-size}")
    private String businessBatchSize;


    @Bean("timeProducerFactory")
    public ProducerFactory<String, String> timeProducerFactory() {
        Map<String, Object> configProps = new HashMap<>();
        configProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        configProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, keySerializer);
        configProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, valueSerializer);
        configProps.put(ProducerConfig.LINGER_MS_CONFIG, timeLingerMs);
        configProps.put(ProducerConfig.BATCH_SIZE_CONFIG, timeBatchSize);
        configProps.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, timeCompressionType);
        configProps.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG, TimePInterceptor.class.getName());
        return new DefaultKafkaProducerFactory<>(configProps);
    }

    @Bean("businessProducerFactory")
    public ProducerFactory<String, String> businessProducerFactory() {
        Map<String, Object> configProps = new HashMap<>();
        configProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        configProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, keySerializer);
        configProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, valueSerializer);
        configProps.put(ProducerConfig.LINGER_MS_CONFIG, businessLingerMs);
        configProps.put(ProducerConfig.BATCH_SIZE_CONFIG, businessBatchSize);
        configProps.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, businessCompressionType);
        configProps.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG, BusinessPInterceptor.class.getName());
        return new DefaultKafkaProducerFactory<>(configProps);
    }

    @Bean("deadProducerFactory")
    public ProducerFactory<String, String> deadProducerFactory() {
        Map<String, Object> configProps = new HashMap<>();
        configProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        configProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, keySerializer);
        configProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, valueSerializer);
        configProps.put(ProducerConfig.LINGER_MS_CONFIG, businessLingerMs);
        configProps.put(ProducerConfig.BATCH_SIZE_CONFIG, businessBatchSize);
        configProps.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, businessCompressionType);
        configProps.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG, BusinessPInterceptor.class.getName());
        return new DefaultKafkaProducerFactory<>(configProps);
    }

    @Bean("timeKafkaTemplate")
    public KafkaTemplate<String, String> timeKafkaTemplate() {
        return new KafkaTemplate<>(timeProducerFactory());
    }

    @Bean("businessKafkaTemplate")
    public KafkaTemplate<String, String> businessKafkaTemplate() {
        return new KafkaTemplate<>(businessProducerFactory());
    }

    @Bean("deadKafkaTemplate")
    public KafkaTemplate<String, String> deadKafkaTemplate() {
        return new KafkaTemplate<>(deadProducerFactory());
    }
}
